package com.tksflysun.im.connect.handle;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.tksflysun.im.common.protobuf.Im;
import com.tksflysun.im.common.protobuf.Im.TcpPackage;
import com.tksflysun.im.connect.task.LoginMsgTask;
import com.tksflysun.im.connect.task.TransferMsgTask;

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * ProtoBuf解码器，约定前四个字节用于指定需要读取的长度
 * 
 * @author liupengfei
 *
 */
@Service
@Sharable
public class MsgHandler extends ChannelInboundHandlerAdapter {
    private static final ExecutorService receiveMsgPool = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<Runnable>(10000), new ThreadPoolExecutor.DiscardPolicy());
    private static final Logger logger = LoggerFactory.getLogger(MsgHandler.class);

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof Im.TcpPackage) {
            Im.TcpPackage tcpPackage = (TcpPackage)msg;
            // 登录消息
            if (tcpPackage.getPackageType() == Im.PkgTypeEnum.LOGIN_REQ_VALUE) {
                receiveMsgPool.execute(new LoginMsgTask(tcpPackage));
                // 心跳消息，直接发送响应
            } else if (tcpPackage.getPackageType() == Im.PkgTypeEnum.PING_VALUE) {
                Im.TcpPackage pong = Im.TcpPackage.newBuilder().setPackageType(Im.PkgTypeEnum.PONG_VALUE).build();
                logger.warn("接收到心跳消息");
                ctx.channel().writeAndFlush(pong);
            } else if (tcpPackage.getPackageType() == Im.PkgTypeEnum.PONG_VALUE) {
                logger.warn("接收到心跳响应消息");
                ctx.fireChannelRead(msg);
            } else {
                receiveMsgPool.execute(new TransferMsgTask(tcpPackage));
            }
        } else {
            ctx.fireChannelRead(msg);
        }

    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        super.channelReadComplete(ctx);
    }
}
